package com.atguigu.flink.exec1;

import com.atguigu.flink.pojo.OrderEvent;
import com.atguigu.flink.pojo.TxEvent;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

/**
 * Created by Smexy on 2023/1/29
 *
 *  思路：
 *          ①读取来自两个数据源的数据，封装为流
 *          ②把流中的一行，封装为数据模型
 *          ③把两种数据(流) Connect合并到一起，进行比对.比对成功，就输出
 *                  多流操作:
 *                      union: 把类型相同的流合并为一个流
 *                      connect: 把类型不同的流，经过处理，输出同类型的数据
 */
public class Demo5_OrderPayCheck
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //①读取来自两个数据源的数据，封装为流 ②把流中的一行，封装为数据模型
        SingleOutputStreamOperator<OrderEvent> orderDs = env.readTextFile("data/OrderLog.csv")
                                                        .map(new MapFunction<String, OrderEvent>()
                                                        {
                                                            @Override
                                                            public OrderEvent map(String value) throws Exception {
                                                                String[] words = value.split(",");

                                                                return new OrderEvent(
                                                                    Long.valueOf(words[0]),
                                                                    words[1],
                                                                    words[2],
                                                                    Long.valueOf(words[3])
                                                                );
                                                            }
                                                        });

        SingleOutputStreamOperator<TxEvent> txDS = env.readTextFile("data/ReceiptLog.csv")
                                                            .map(new MapFunction<String, TxEvent>()
                                                            {
                                                                @Override
                                                                public TxEvent map(String value) throws Exception {
                                                                    String[] words = value.split(",");

                                                                    return new TxEvent(
                                                                        words[0],
                                                                        words[1],
                                                                        Long.valueOf(words[2])
                                                                    );
                                                                }
                                                            });

        //③把两种数据(流) Connect合并到一起，进行比对.比对成功，就输出
        ConnectedStreams<OrderEvent, TxEvent> checkDS = orderDs.connect(txDS);

        checkDS
            //ConnectedStreams 调用keyBy，需要传两个参数。 保证txId相同的数据一定是发到下游的同一个process算子中处理，才有对账成功的可能！
            .keyBy(OrderEvent::getTxId,TxEvent::getTxId)
            .process(new CoProcessFunction<OrderEvent, TxEvent, Long>()
        {

            //准备两个缓存 ①早到的数据，先缓存起来，等候后续到达的数据处理  ②作为属性存在，可以方便两个方法相互交换变量
            //对于OrderEvent 要缓存: orderId ,txId。 二者是1:1的关系
            Map<String,Long> orderCache = new HashMap<>();
            Set<String> txCache = new HashSet<>();
            //对于TxEvent 要缓存: txId

            //处理第一个流  orderDs
            @Override
            public void processElement1(OrderEvent oe, Context ctx, Collector<Long> out) throws Exception {

                //到txEvent的缓存中匹配，成功就输出，找不到，就把oe写入OrderCache
                if (txCache.contains(oe.getTxId())){
                    out.collect(oe.getOrderId());
                }else{
                    orderCache.put(oe.getTxId(),oe.getOrderId());
                }


            }

            //处理第二个流 txDS
            @Override
            public void processElement2(TxEvent te, Context ctx, Collector<Long> out) throws Exception {
                //到orderCache的缓存中匹配，成功就输出，找不到，就把te写入txCache
                Long orderId = orderCache.get(te.getTxId());
                if(orderId == null){
                    txCache.add(te.getTxId());
                }else{
                    out.collect(orderId);
                }

            }
        })
            .print().setParallelism(1);


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
